package com.nanohadoop.mapreduce.core;

import java.io.IOException;
import java.util.Iterator;

/**
 * Reducer 接口定义了 MapReduce 框架中 reduce 阶段的基本操作
 * 用户通过继承该类并实现 reduce 方法来定义具体的 reduce 逻辑
 * 
 * 支持键值对遍历
支持进度报告
提供了可扩展的接口
 */
public abstract class Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {

    private ReducerContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> context;

    /**
     * 设置运行上下文
     */
    public void setContext(ReducerContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> context) {
        this.context = context;
    }

    /**
     * 初始化方法,在reduce任务开始前调用
     */
    protected void setup(ReducerContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> context) throws IOException, InterruptedException {
        // 默认空实现,用户可以重写此方法进行初始化操作
    }

    /**
     * reduce方法,处理一组具有相同key的values
     * @param key 输入key
     * @param values 相同key对应的所有value的迭代器
     * @param context 用于输出结果的上下文
     */
    public abstract void reduce(KEYIN key, Iterator<VALUEIN> values,
                                 ReducerContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> context) throws IOException, InterruptedException;

    /**
     * 清理方法,在reduce任务结束后调用
     */
    protected void cleanup(ReducerContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> context) throws IOException, InterruptedException {
        // 默认空实现,用户可以重写此方法进行清理操作
    }

    /**
     * run方法封装了reduce任务的执行流程
     */
    public void run(ReducerContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> context) throws IOException, InterruptedException {
        setup(context);
        try {
            while (context.nextKey()) {
                reduce(context.getCurrentKey(), context.getValues(), context);
            }
        } finally {
            cleanup(context);
        }
    }

}